package com.zwcdcc.coprocessor;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

public class UrrDelObs extends BaseRegionObserver {
    private TableName tableName = TableName.valueOf("user_relate");
    private byte[] family_friends = Bytes.toBytes("friends");

    public boolean not_exist(Table table, byte[] row, byte[] qualifier) throws IOException {
        final Get get = new Get(row);
        get.addColumn(family_friends, qualifier);
        return table.exists(get) ? false : true;
    }

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        final Table table = e.getEnvironment().getTable(tableName);
        final List<Cell> cells = delete.getFamilyCellMap().get(family_friends);
        for (Cell cell : cells) {
            byte[] cloneQualifier = CellUtil.cloneQualifier(cell);
            byte[] cloneRow = CellUtil.cloneRow(cell);
            // 判断删除的好友及其好友列表中是否存在该用户，不存在则放行
            if (not_exist(table, cloneQualifier, cloneRow)) continue;
            // 删除关联数据
            Delete del = new Delete(cloneQualifier);
            del.addColumn(family_friends, cloneRow);
            table.delete(del);
        }
        table.close();
    }
}
